package com.owen.flink.batch.java.relational;

import com.owen.flink.batch.java.relational.util.WebLogData;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.functions.FunctionAnnotation;
import org.apache.flink.api.java.operators.FilterOperator;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.util.Collector;

/**
 * This program processes web logs and relational data.
 * It implements the following relational query:
 *
 * <pre>{@code
 * SELECT
 *       r.pageURL,
 *       r.pageRank,
 *       r.avgDuration
 * FROM documents d JOIN rankings r
 *                  ON d.url = r.url
 * WHERE CONTAINS(d.text, [keywords])
 *       AND r.rank > [rank]
 *       AND NOT EXISTS
 *           (
 *              SELECT * FROM Visits v
 *              WHERE v.destUrl = d.url
 *                    AND v.visitDate < [date]
 *           );
 * }</pre>
 *
 * <p>Input files are plain text CSV files using the pipe character ('|') as field separator.
 * The tables referenced in the query can be generated using the {@link com.owen.flink.batch.java.relational.util.WebLogDataGenerator} and
 * have the following schemas
 * <pre>{@code
 * CREATE TABLE Documents (
 *                url VARCHAR(100) PRIMARY KEY,
 *                contents TEXT );
 *
 * CREATE TABLE Rankings (
 *                pageRank INT,
 *                pageURL VARCHAR(100) PRIMARY KEY,
 *                avgDuration INT );
 *
 * CREATE TABLE Visits (
 *                sourceIP VARCHAR(16),
 *                destURL VARCHAR(100),
 *                visitDate DATE,
 *                adRevenue FLOAT,
 *                userAgent VARCHAR(64),
 *                countryCode VARCHAR(3),
 *                languageCode VARCHAR(6),
 *                searchWord VARCHAR(32),
 *                duration INT );
 * }</pre>
 *
 * <p>Usage: <code>WebLogAnalysis --documents &lt;path&gt; --ranks &lt;path&gt; --visits &lt;path&gt; --result &lt;path&gt;</code><br>
 * If no parameters are provided, the program is run with default data from {@link com.owen.flink.batch.java.relational.util.WebLogData}.
 *
 * <p>This example shows how to use:
 * <ul>
 * <li> tuple data types
 * <li> projection and join projection
 * <li> the CoGroup transformation for an anti-join
 * </ul>
 */
public class WebLogAnalysis {

    public static void main(String[] args) {

        final ParameterTool params = ParameterTool.fromArgs(args);

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        env.getConfig().setGlobalJobParameters(params);

        DataSet<Tuple2<String, String>> documentDataSet = WebLogData.getDocumentDataSet(env);
        DataSet<Tuple3<Integer, String, Integer>> rankDataSet = WebLogData.getRankDataSet(env);
        DataSet<Tuple2<String, String>> visitDataSet = WebLogData.getVisitDataSet(env);

        DataSet<Tuple1<String>> filterDocs = documentDataSet.filter(new KeywordsFilter()).project(0);

        DataSet<Tuple3<Integer, String, Integer>> filterRanks = rankDataSet.filter(new RankFilter());

        DataSet<Tuple2<String, String>> filterVisit = visitDataSet.filter(new VisitFilter());

        DataSet<Tuple3<Integer, String, Integer>> joinDocsRanks=filterDocs.join(filterRanks).where(0).equalTo(1).projectSecond(0,1,2);

//        joinDocsRanks.coGroup(filterVisit).where(1).equalTo(0).with(   );

//        visitDataSet.filter(new VisitFilter()).join(filterDocs).where(0);

    }

    private static class KeywordsFilter implements FilterFunction<Tuple2<String, String>> {
        private static final String[] KEYWORDS = {" editors ", " oscillations "};

        @Override
        public boolean filter(Tuple2<String, String> value) throws Exception {
//            List<String> keywords=Lists.newArrayList("dolor");
//            List<String> data=Splitter.on(" ").splitToList(value.f1);
//            for (int i = 0,size=keywords.size(); i <size ; i++) {
//                if(data.contains(keywords.get(i))){
//                    return true;
//                }
//            }
//            return false;
            // FILTER
            // Only collect the document if all keywords are contained
            String docText = value.f1;
            for (String kw : KEYWORDS) {
                if (!docText.contains(kw)) {
                    return false;
                }
            }
            return true;
        }
    }

    private static class RankFilter implements FilterFunction<Tuple3<Integer, String, Integer>> {
        private static final Integer rank=10;

        @Override
        public boolean filter(Tuple3<Integer, String, Integer> value) throws Exception {
            return value.f0>rank;
        }
    }

    private static class VisitFilter implements FilterFunction<Tuple2<String, String>> {
        private static final int YEARFILTER = 2007;

        @Override
        public boolean filter(Tuple2<String, String> value) throws Exception {
            String dateString = value.f1;
            int year = Integer.parseInt(dateString.substring(0, 4));
            return year==YEARFILTER;
        }
    }

    /**
     * CoGroupFunction that realizes an anti-join.
     * If the first input does not provide any pairs, all pairs of the second input are emitted.
     * Otherwise, no pair is emitted.
     */
    @FunctionAnnotation.ForwardedFieldsFirst("*")
    public static class AntiJoinVisits implements CoGroupFunction<Tuple3<Integer, String, Integer>, Tuple1<String>, Tuple3<Integer, String, Integer>> {

        /**
         * If the visit iterator is empty, all pairs of the rank iterator are emitted.
         * Otherwise, no pair is emitted.
         *
         * <p>Output Format:
         * 0: RANK
         * 1: URL
         * 2: AVG_DURATION
         */
        @Override
        public void coGroup(Iterable<Tuple3<Integer, String, Integer>> ranks, Iterable<Tuple1<String>> visits, Collector<Tuple3<Integer, String, Integer>> out) {
            // Check if there is a entry in the visits relation
            if (!visits.iterator().hasNext()) {
                for (Tuple3<Integer, String, Integer> next : ranks) {
                    // Emit all rank pairs
                    out.collect(next);
                }
            }
        }
    }
}
